package com.changdu.seatunnel.admin.service;

import java.time.Duration;
import java.time.LocalDateTime;
import java.util.List;
import java.util.Objects;

import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;

import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.baomidou.mybatisplus.extension.service.impl.ServiceImpl;
import com.changdu.seatunnel.admin.config.enums.StartModeEnum;
import com.changdu.seatunnel.admin.config.enums.SyncTaskStatus;
import com.changdu.seatunnel.admin.config.enums.SyncType;
import com.changdu.seatunnel.admin.entity.DataKafkaTemplate;
import com.changdu.seatunnel.admin.entity.SyncTask;
import com.changdu.seatunnel.admin.mapper.SyncTaskMapper;
import com.changdu.seatunnel.admin.pojo.JobParams;
import com.changdu.seatunnel.admin.pojo.PageSyncTasks;
import com.changdu.seatunnel.admin.pojo.SeatunnelTaskInfo;
import com.changdu.seatunnel.admin.util.JsonUtils;
import com.changdu.seatunnel.admin.util.KafkaToStarrocksTempltes;
import com.changdu.seatunnel.admin.util.MysqlToKafkaTempltes;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;

@Service
public class SyncTaskService extends ServiceImpl<SyncTaskMapper, SyncTask> {

	@Autowired
	private SeatunnelRester seatunnelRester;
	
	@Autowired
	private DataKafkaTemplateService dataKafkaTemplateService;	
	
	@Autowired
	private KafkaCreater kafkaClient;
	
	/**
	 * 提交任务
	 * @param id
	 * @throws InterruptedException 
	 */
	public void submitSeatunnelJob(String id, JobParams jobParam) throws InterruptedException {
		SyncTask task = getById(id);
		if (Objects.isNull(task)) throw new IllegalArgumentException("任务不存在");
		
		SyncType syncType = SyncType.fromCode(task.getSourcePlugin(), task.getSinkPlugin());
		if (Objects.nonNull(jobParam)) {
			// 模板变更后，需要暂停历史任务，再提交, 此时需要重新更新任务ID
			boolean modifyStartMode = modifySeatunnelJobWithParam(task, syncType, jobParam);
			if (modifyStartMode) {
				seatunnelRester.sendStopRequest(task.getDomain(), task.getTaskId(), Boolean.FALSE);
				Thread.sleep(Duration.ofSeconds(2));
				task.setTaskId(null);
			}
		}
		if (syncType.isKafkaSource()) {
			// 如果任务是 kafka 消费任务，并且存在子任务模板，先提交子任务模板
			List<DataKafkaTemplate> schemaTemplates = dataKafkaTemplateService.getTemplatesBySyncConfigId(task.getSyncConfigId());
			for (DataKafkaTemplate dataKafkaTemplate : schemaTemplates) {
				seatunnelRester.sendKafksSubRequest(task.getDomain(), dataKafkaTemplate.getTemplateContent());
			}
		}

		String taskId = seatunnelRester.sendJobRequest(task.getDomain(), task.getTaskId(), task.getTemplateContent());
		task.setTaskId(taskId);
		task.setStatus(SyncTaskStatus.IN_PROGRESS.getCode());
		updateByIdWithUpdateTime(task);
	}

	/**
	 * 依据任务参数调整模板
	 * @param id
	 * @param jobParam
	 */
	private boolean modifySeatunnelJobWithParam(SyncTask task, SyncType syncType, JobParams jobParam) {
		if (jobParam.getConsumeMode() == StartModeEnum.NONE) return false;
		
		if(syncType.getSource().equals(jobParam.getType())) {
			switch (syncType) {
				case KAFKA_STARROCKS: {
					task.setTemplateContent(KafkaToStarrocksTempltes.reCreateTemplate(task.getTemplateContent(), jobParam));
					return true;
				}
				case MYSQL_KAFKA: {
					task.setTemplateContent(MysqlToKafkaTempltes.reCreateTemplate(task.getTemplateContent(), jobParam));
					return true;
				}
				default: break;
			}
		}
		return false;
	}
	
	/**
	 * 停止任务
	 * @param id
	 */
	public void stopSeatunnelJob(String id, Boolean savepoint) {
		SyncTask task = getById(id);
		if (Objects.isNull(task)) throw new IllegalArgumentException("任务不存在");
		
		seatunnelRester.sendStopRequest(task.getDomain(), task.getTaskId(), savepoint);
		task.setStatus(SyncTaskStatus.STOP.getCode());
		updateByIdWithUpdateTime(task);
	}

	public boolean updateByIdWithUpdateTime(SyncTask task) {
		task.setUpdateTime(LocalDateTime.now().toString());
		return updateById(task);
	}
	
	/**
	 * 删除任务
	 * @param id
	 * @return 
	 */
	public boolean deleteSeatunnelJob(Integer id) {
		SyncTask task = getById(id);
		if (Objects.isNull(task)) throw new IllegalArgumentException("任务不存在");
		
		if (SyncTaskStatus.IN_PROGRESS == SyncTaskStatus.fromCode(task.getStatus())) {
			seatunnelRester.sendDeleteRequest(task.getDomain(), task.getTaskId());
		}
		return removeById(id);
	}

	/**
	 * 更新子模板内容
	 * @param id
	 */
	public void submitSeatunnelSubJob(Integer id) {
		DataKafkaTemplate schemaTemplate = dataKafkaTemplateService.getById(id);
		seatunnelRester.sendKafksSubRequest(schemaTemplate.getDomain(), schemaTemplate.getTemplateContent());
	}

	/**
	 * 创建任务相关 topic
	 * @param id
	 */
	public void createSyncTaskTopic(Integer id) {
		SyncTask syncTask = getById(id);
		SyncType syncType = SyncType.fromCode(syncTask.getSourcePlugin(), syncTask.getSinkPlugin());
		JsonNode template = JsonUtils.getJsonNode(syncTask.getTemplateContent().getBytes());
		
		// 从模板中提取kafka.conf相关参数，检查并创建topic
		if (syncType == SyncType.MYSQL_KAFKA || syncType == SyncType.TIDB_KAFKA) {
			JsonNode sink = template.get("sink").get(0);
			kafkaClient.createTopicIfNotExists(sink);
		}
		
		if (syncType == SyncType.KAFKA_STARROCKS) {
			JsonNode source = template.get("source").get(0);
			kafkaClient.createTopicIfNotExists(source);
		}
	}

	/**
	 * 更新任务名称，同时更新模板名称
	 * @param syncTask
	 * @return
	 */
	public boolean updateTaskName(SyncTask syncTask) {
		SyncTask task = getById(syncTask.getId());
		// 更新名称
		task.setShowName(syncTask.getShowName());
		// 更新模板名称
		JsonNode template = JsonUtils.getJsonNode(task.getTemplateContent().getBytes());
		((ObjectNode) template.get("env")).put("job.name", syncTask.getShowName());
		task.setTemplateContent(JsonUtils.toJson(template));
		
		return updateByIdWithUpdateTime(task);
	}

	/**
	 * 查询所有运行中的任务
	 * @throws InterruptedException 
	 */
	public List<SyncTask> getRunningTasks() throws InterruptedException {
		LambdaQueryWrapper<SyncTask> query = new LambdaQueryWrapper<SyncTask>()
				.eq(SyncTask::getStatus, SyncTaskStatus.IN_PROGRESS.getCode())
				.isNotNull(SyncTask::getTaskId);
		return list(query);
	}

	/**
	 * 根据配置ID列表查询同步任务
	 *
	 * @param configIds 配置ID列表
	 * @param current   当前页码
	 * @param size      每页大小
	 * @return 同步任务列表
	 */
	public PageSyncTasks listBySyncConfigIds(List<Long> configIds, Page<SyncTask> pageWrap) {
		if (configIds == null || configIds.isEmpty()) return PageSyncTasks.create(List.of(), 0L);
		
		LambdaQueryWrapper<SyncTask> query = new LambdaQueryWrapper<SyncTask>().in(SyncTask::getSyncConfigId, configIds)
			.orderByDesc(SyncTask::getSyncConfigId, SyncTask::getUpdateTime);
		return PageSyncTasks.create(list(pageWrap, query), count(query));
	}	
	
	/**
	 * 根据任务类型及任务名称进行搜索
	 * @param pageWrap
	 * @param jobName
	 * @param syncType
	 * @return
	 */
	public PageSyncTasks listByJobNameOrSyncType(Page<SyncTask> pageWrap, String jobName, String syncType) {
		LambdaQueryWrapper<SyncTask> query = new LambdaQueryWrapper<SyncTask>();
		if (StringUtils.isNotBlank(jobName)) {
			query.like(StringUtils.isNotBlank(jobName), SyncTask::getName, jobName);
		}
		if (StringUtils.isNotBlank(syncType)) {
			query.eq(StringUtils.isNotBlank(syncType), SyncTask::getTaskType, Integer.valueOf(syncType));
		}
		query.orderByDesc(SyncTask::getSyncConfigId, SyncTask::getUpdateTime);
		return PageSyncTasks.create(list(pageWrap, query), count(query));
	}

	public PageSyncTasks listOrderByUpdateTime(Page<SyncTask> pageWrap) {
		LambdaQueryWrapper<SyncTask> query = new LambdaQueryWrapper<SyncTask>();
		query.orderByDesc(SyncTask::getSyncConfigId, SyncTask::getUpdateTime);
		return PageSyncTasks.create(list(pageWrap, query), count(query));
	}

	/**
	 * 按任务ID搜索
	 * @param pageWrap
	 * @param stTaskId
	 * @return
	 */
	public PageSyncTasks getByStTaskId(Page<SyncTask> pageWrap, String stTaskId) {
		LambdaQueryWrapper<SyncTask> query = new LambdaQueryWrapper<SyncTask>();
		query.eq(SyncTask::getTaskId, stTaskId);
		return PageSyncTasks.create(list(pageWrap, query), count(query));
	}

	/**
	 * 同步任务状态
	 * @param syncTask
	 * @return
	 */
	public SeatunnelTaskInfo syncTaskState(SyncTask syncTask) {
		SeatunnelTaskInfo taskStatus = seatunnelRester.getSyncTaskStatus(syncTask.getDomain(), syncTask.getTaskId());
		syncTask.setStatus(taskStatus.taskStatus().getCode());
		updateById(syncTask);
		return taskStatus;
	}

}